package com.wish233.register.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.wish233.config.RegistryConfig;
import com.wish233.model.register.ServiceMetaInfo;
import com.wish233.register.Registry;
import com.wish233.utils.ServiceMetaUtils;
import io.vertx.core.impl.ConcurrentHashSet;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;

import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author WHH
 */
@Slf4j
public class ZookeeperRegistry implements Registry {

    public static final String ZK_ROOT_PATH = "/wrpc";
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();
    private CuratorFramework client;
    private RegistryServiceCache cache;
    private ServiceDiscovery<ServiceMetaInfo> serviceDiscovery;

    @Override
    public void init(RegistryConfig registryConfig) {
        String address = registryConfig.getAddress();
        Long timeout = registryConfig.getTimeout();
        cache = RegistryServiceCache.getCache();
        //质数退避连接策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(Math.toIntExact(timeout), 3);
        client = CuratorFrameworkFactory.builder()
                //连接地址
                .connectString(address)
                //设置会话timeout
                .sessionTimeoutMs(10000)
                .connectionTimeoutMs(Math.toIntExact(timeout))
                //重试策略
                .retryPolicy(retryPolicy)
                .build();
        serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMetaInfo.class)
                .basePath(ZK_ROOT_PATH)
                .client(client)
                .serializer(new JsonInstanceSerializer<>(ServiceMetaInfo.class))
                .build();
        try {
            client.start();
            serviceDiscovery.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        //添加到zk
        serviceDiscovery.registerService(buildServiceInstance(serviceMetaInfo));
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        try {
            serviceDiscovery.unregisterService(buildServiceInstance(serviceMetaInfo));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        //从缓存获取服务
        List<ServiceMetaInfo> serviceMetaInfos = cache.readCache(serviceKey);
        if (CollectionUtil.isNotEmpty(serviceMetaInfos)) {
            return serviceMetaInfos;
        } else {
            //缓存为空 从zk查
            try {
                Collection<ServiceInstance<ServiceMetaInfo>> serviceInstances
                        = serviceDiscovery.queryForInstances(serviceKey);
                //获取服务信息
                //监听
                List<ServiceMetaInfo> metaInfoList =
                        serviceInstances.stream()
                                .map(ServiceInstance::getPayload)
                                .collect(Collectors.toList());
                //写缓存
                cache.writeCache(serviceKey, metaInfoList);
                //监听
                for (ServiceMetaInfo info : metaInfoList) {
                    watch(info.getServiceKey());
                }
                return metaInfoList;
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            //写缓存
        }
    }

    @Override
    public void heartBeat() {
        //创建的是临时节点，无需心跳
    }

    @Override
    public void destroy() {
        //临时节点，服务下线自动滚
        if (client != null) {
            client.close();
        }
    }

    @Override
    public void watch(String serviceKey) {
        if (StrUtil.isBlank(serviceKey)) {
            return;
        }
        String regKey = ZK_ROOT_PATH + "/" + serviceKey;
        if (!watchingKeySet.add(regKey)) {
            return;
        }
        // 推荐方式：监听目录节点的子节点变化（例如 /user-service:1.0）
        PathChildrenCache childrenCache = new PathChildrenCache(client, regKey, true);
        try {
            childrenCache.start();
            childrenCache.getListenable().addListener((client, event) -> {
                switch (event.getType()) {
                    case CHILD_ADDED:
                    case CHILD_UPDATED:
                    case CHILD_REMOVED:
                        log.info("服务节点变动：{}，清除缓存", event.getData().getPath());
                        cache.clearCache(serviceKey);
                        break;
                    default:
                        break;
                }
            });
        } catch (Exception e) {
            throw new RuntimeException("监听失败：" + serviceKey, e);
        }
    }



    private ServiceInstance<ServiceMetaInfo> buildServiceInstance(ServiceMetaInfo serviceMetaInfo) {
        String serviceAddress = serviceMetaInfo.getServiceAddress();
        try {
            return ServiceInstance.<ServiceMetaInfo>builder()
                    .address(serviceAddress)
                    .id(serviceAddress)
                    .name(serviceMetaInfo.getServiceKey())
                    .payload(serviceMetaInfo)
                    .build();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
